package drds.binlog.parse.mysql.table_meta_data;

import com.alibaba.fastsql.sql.ast.SQLDataType;
import com.alibaba.fastsql.sql.ast.SQLDataTypeImpl;
import com.alibaba.fastsql.sql.ast.SQLExpr;
import com.alibaba.fastsql.sql.ast.SQLStatement;
import com.alibaba.fastsql.sql.ast.expr.*;
import com.alibaba.fastsql.sql.ast.statement.*;
import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlPrimaryKey;
import com.alibaba.fastsql.sql.dialect.mysql.ast.MySqlUnique;
import com.alibaba.fastsql.sql.repository.Schema;
import com.alibaba.fastsql.sql.repository.SchemaObject;
import com.alibaba.fastsql.sql.repository.SchemaRepository;
import com.alibaba.fastsql.util.JdbcConstants;
import drds.binlog.common.Authors;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.parse.FieldMetaData;
import drds.binlog.parse.TableMetaData;
import drds.binlog.parse.mysql.ddl.FastsqlDdlParser;
import drds.common.$;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 基于DDL维护的内存表结构
 */
@Slf4j
@Author(name = Authors.LI_YANG)
public class BaseOnMemoryTableMetaDataGetter implements TableMetaDataGetter
{
    @Setter
    @Getter
    private Map<List<String>, TableMetaData> schemaNameAndTableNameListToTableMetaDataMap = new ConcurrentHashMap<List<String>, TableMetaData>();
    @Setter
    @Getter
    private SchemaRepository schemaRepository = new SchemaRepository(JdbcConstants.MYSQL);

    public BaseOnMemoryTableMetaDataGetter()
    {
    }

    @Override
    public boolean init(String destination)
    {
        return true;
    }

    public boolean apply(EntryPosition entryPosition, String schemaName, String ddl, String extra)
    {
        schemaNameAndTableNameListToTableMetaDataMap.clear();
        synchronized (this)
        {
            if ($.isNotNullAndNotEmpty(schemaName))
            {
                schemaRepository.setDefaultSchema(schemaName);
            }

            try
            {
                // druid暂时flush privileges语法解析有问题
                if (!StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "flush")
                        && !StringUtils.startsWithIgnoreCase(StringUtils.trim(ddl), "grant"))
                {
                    schemaRepository.console(ddl);
                }
            } catch (Throwable e)
            {
                log.warn("parse faield : " + ddl, e);
            }
        }


        return true;
    }

    @Override
    public TableMetaData find(String schemaName, String tableName)
    {
        List<String> schemaNameAndTableNameList = Arrays.asList(schemaName, tableName);
        TableMetaData tableMetaData = schemaNameAndTableNameListToTableMetaDataMap.get(schemaNameAndTableNameList);

        if (tableMetaData != null)
        {
            return tableMetaData;
        } else
        {
            synchronized (this)
            {
                tableMetaData = schemaNameAndTableNameListToTableMetaDataMap.get(schemaNameAndTableNameList);
                if (tableMetaData == null)
                {
                    Schema schema = schemaRepository.findSchema(schemaName);
                    if (schema == null)
                    {
                        return null;
                    }
                    SchemaObject schemaTable = schema.findTable(tableName);
                    if (schemaTable == null)
                    {
                        return null;
                    }
                    SQLStatement statement = schemaTable.getStatement();
                    if (statement == null)
                    {
                        return null;
                    }
                    if (statement instanceof SQLCreateTableStatement)
                    {
                        tableMetaData = parse((SQLCreateTableStatement) statement);
                    }
                    if (tableMetaData != null)
                    {
                        if (tableName != null)
                        {
                            tableMetaData.setTableName(tableName);
                        }
                        if (schemaName != null)
                        {
                            tableMetaData.setDatabaseName(schemaName);
                        }

                        schemaNameAndTableNameListToTableMetaDataMap.put(schemaNameAndTableNameList, tableMetaData);
                    }
                }
            }
            return tableMetaData;
        }


    }

    @Override
    public boolean rollback(EntryPosition entryPosition)
    {
        throw new RuntimeException("not support for memory");
    }

    public Map<String, String> snapshot()
    {
        Map<String, String> schemaNameToDdlMap = new HashMap<String, String>();
        for (Schema schema : schemaRepository.getSchemas())
        {
            StringBuffer sb = new StringBuffer(4 * 1024);
            for (String tableName : schema.showTables())
            {
                SchemaObject schemaObject = schema.findTable(tableName);
                schemaObject.getStatement().output(sb);
                sb.append("; \n");
            }
            schemaNameToDdlMap.put(schema.getName(), sb.toString());
        }

        return schemaNameToDdlMap;
    }

    private TableMetaData parse(SQLCreateTableStatement sqlCreateTableStatement)
    {
        int size = sqlCreateTableStatement.getTableElementList().size();
        if (size > 0)
        {
            TableMetaData tableMetaData = new TableMetaData();
            for (int i = 0; i < size; ++i)
            {
                SQLTableElement sqlTableElement = sqlCreateTableStatement.getTableElementList().get(i);
                processTableElement(sqlTableElement, tableMetaData);
            }
            return tableMetaData;
        }

        return null;
    }

    private void processTableElement(SQLTableElement sqlTableElement, TableMetaData tableMetaData)
    {
        if (sqlTableElement instanceof SQLColumnDefinition)
        {
            FieldMetaData fieldMetaData = new FieldMetaData();
            SQLColumnDefinition sqlColumnDefinition = (SQLColumnDefinition) sqlTableElement;
            String columnName = getSqlName(sqlColumnDefinition.getName());
            // String connectionCharset = getSqlName(column.getCharsetExpr());
            SQLDataType sqlDataType = sqlColumnDefinition.getDataType();
            String columnType = sqlDataType.getName();
            if (sqlDataType.getArguments().size() > 0)
            {
                columnType += "(";
                for (int i = 0; i < sqlColumnDefinition.getDataType().getArguments().size(); i++)
                {
                    if (i != 0)
                    {
                        columnType += ",";
                    }
                    SQLExpr arg = sqlColumnDefinition.getDataType().getArguments().get(i);
                    columnType += arg.toString();
                }
                columnType += ")";
            }

            if (sqlDataType instanceof SQLDataTypeImpl)
            {
                SQLDataTypeImpl sqlDataType1 = (SQLDataTypeImpl) sqlDataType;
                if (sqlDataType1.isUnsigned())
                {
                    columnType += " unsigned";
                }

                if (sqlDataType1.isZerofill())
                {
                    columnType += " zerofill";
                }
            }

            if (sqlColumnDefinition.getDefaultExpr() == null || sqlColumnDefinition.getDefaultExpr() instanceof SQLNullExpr)
            {
                fieldMetaData.setDefaultValue(null);
            } else
            {
                fieldMetaData.setDefaultValue(FastsqlDdlParser.unescapeQuotaName(getSqlName(sqlColumnDefinition.getDefaultExpr())));
            }

            fieldMetaData.setColumnName(columnName);
            fieldMetaData.setColumnType(columnType);
            fieldMetaData.setNullable(true);
            List<SQLColumnConstraint> sqlColumnConstraintList = sqlColumnDefinition.getConstraints();
            for (SQLColumnConstraint sqlColumnConstraint : sqlColumnConstraintList)
            {
                if (sqlColumnConstraint instanceof SQLNotNullConstraint)
                {
                    fieldMetaData.setNullable(false);
                } else if (sqlColumnConstraint instanceof SQLNullConstraint)
                {
                    fieldMetaData.setNullable(true);
                } else if (sqlColumnConstraint instanceof SQLColumnPrimaryKey)
                {
                    fieldMetaData.setPrimaryKey(true);
                } else if (sqlColumnConstraint instanceof SQLColumnUniqueKey)
                {
                    fieldMetaData.setUniqueKey(true);
                }
            }
            tableMetaData.getFieldMetaDataList().add(fieldMetaData);
        } else if (sqlTableElement instanceof MySqlPrimaryKey)
        {
            MySqlPrimaryKey mySqlPrimaryKey = (MySqlPrimaryKey) sqlTableElement;
            List<SQLSelectOrderByItem> sqlSelectOrderByItemList = mySqlPrimaryKey.getColumns();
            for (SQLSelectOrderByItem sqlSelectOrderByItem : sqlSelectOrderByItemList)
            {
                String columnName = getSqlName(sqlSelectOrderByItem.getExpr());
                FieldMetaData fieldMetaData = tableMetaData.getFieldMetaData(columnName);
                fieldMetaData.setPrimaryKey(true);
            }
        } else if (sqlTableElement instanceof MySqlUnique)
        {
            MySqlUnique mySqlUnique = (MySqlUnique) sqlTableElement;
            List<SQLSelectOrderByItem> sqlSelectOrderByItemList = mySqlUnique.getColumns();
            for (SQLSelectOrderByItem sqlSelectOrderByItem : sqlSelectOrderByItemList)
            {
                String columnName = getSqlName(sqlSelectOrderByItem.getExpr());
                FieldMetaData fieldMetaData = tableMetaData.getFieldMetaData(columnName);
                fieldMetaData.setUniqueKey(true);
            }
        }
    }

    private String getSqlName(SQLExpr sqlExpr)
    {
        if (sqlExpr == null)
        {
            return null;
        }

        if (sqlExpr instanceof SQLPropertyExpr)
        {
            SQLIdentifierExpr sqlIdentifierExpr = (SQLIdentifierExpr) ((SQLPropertyExpr) sqlExpr).getOwner();
            return FastsqlDdlParser.unescapeName(sqlIdentifierExpr.getName()) + "."
                    + FastsqlDdlParser.unescapeName(((SQLPropertyExpr) sqlExpr).getName());
        } else if (sqlExpr instanceof SQLIdentifierExpr)
        {
            return FastsqlDdlParser.unescapeName(((SQLIdentifierExpr) sqlExpr).getName());
        } else if (sqlExpr instanceof SQLCharExpr)
        {
            return ((SQLCharExpr) sqlExpr).getText();
        } else if (sqlExpr instanceof SQLMethodInvokeExpr)
        {
            return FastsqlDdlParser.unescapeName(((SQLMethodInvokeExpr) sqlExpr).getMethodName());
        } else
        {
            return sqlExpr.toString();
        }
    }


}
